package com.asap.demo.rete;


import com.asap.demo.ContextInfo;
import com.asap.demo.function.dealMapFunction;
import com.asap.demo.function.dealStreamProcessFunction;
import com.asap.demo.model.BeanField;
import com.asap.demo.sourcefunc.MysqlSourceFunction1;
import com.asap.demo.table.JsonField;
import com.asap.demo.table.RuleParse;
import com.asap.demo.utils.Constants;
import com.asap.demo.utils.DateUtil;
import com.asap.demo.utils.Utils;
import com.asap.interf.Action;
import com.asap.rule.StandardEvent;
import com.asap.rule.engine.InferenceEngine;
import com.asap.rule.engine.PatternMatcher;
import com.asap.rule.orm.DbFetcher;
import com.asap.rule.util.PropTransformMap;
import com.asap.rule.util.RuleReader;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.ListTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.io.IOException;
import java.sql.Timestamp;
import java.sql.Types;
import java.time.Instant;
import java.util.*;

public class ReteDemo4 {

	private static final Logger logger = LoggerFactory.getLogger(ReteDemo4.class);
//3085654
	//kafka-run-class kafka.tools.ConsumerOffsetChecker --zookeeper 10.28.184.25:1813 --group temporal --topic flink_pressure_test6

	public static void main(String[] args) throws Exception {

		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
				.useBlinkPlanner()
				.inStreamingMode()
				.build();
		StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);


		//env.setParallelism(3);
		env.enableCheckpointing(5000);  //检查点 每5000ms
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

		Properties browseProperties = new Properties();
		browseProperties.put("bootstrap.servers", "10.28.184.25:9093");
		browseProperties.put("group.id", "temporal");
		browseProperties.put("auto.offset.reset", "latest");
		PropTransformMap.getInstance().readConfigMap("/home/asap/wbh/conf/cfg.properties");
		Map<String, String> configMap = new HashMap<String, String>();
		configMap.put(Constants.DB_JDBC_USER, "root");
		configMap.put(Constants.DB_JDBC_PASSWD, "1qazXSW@3edc");
		configMap.put(Constants.DB_JDBC_URL, "jdbc:mysql://10.28.184.25:3306/SSA?useUnicode=true&characterEncoding=utf-8");
		configMap.put(Constants.DB_JDBC_DRIVER, "com.mysql.jdbc.Driver");
		configMap.put(Constants.INITAL_POOL_SIZE, "10");
		configMap.put(Constants.MIN_POOL_SIZE, "5");
		configMap.put(Constants.MAX_IDLE_TIME, "50");
		configMap.put(Constants.MAX_STATE_ELEMENTS, "100");
		configMap.put(Constants.MAX_IDLE_TIME, "60");
		DbFetcher dbFetcher = new DbFetcher(configMap);
		List<String> listRule = RuleReader.readRules(dbFetcher);
		System.out.println("ListRule::" + listRule.size());
		RuleParse ruleParse = new RuleParse();
		Map properties = new HashMap();
		ruleParse.parseData("/home/asap/wbh/conf/cfg.json");
		logger.info("1111size:" + ruleParse.getTableDefine().getJsonFieldList());
		//1、读取mysql的配置消息
		DataStream<List<String>> conf = env.addSource(new MysqlSourceFunction1(dbFetcher));

		//2、创建MapStateDescriptor规则，对广播的数据的数据类型的规则
		MapStateDescriptor<String, List<String>> ruleStateDescriptor = new MapStateDescriptor<>(ContextInfo.RULE_SBROAD_CAST_STATE
				, BasicTypeInfo.STRING_TYPE_INFO
				, new ListTypeInfo<>(String.class));
		//3、对conf进行broadcast返回BroadcastStream
		final BroadcastStream<List<String>> confBroadcast = conf.broadcast(ruleStateDescriptor);

		DataStream<String> dataStream = env
				.addSource(new FlinkKafkaConsumer<>(
						"flink_pressure_test6",
						new SimpleStringSchema(),
						browseProperties
				));

		DataStream<StandardEvent> kafkaData = dataStream
				.map(new MapFunction<String, StandardEvent>() {
					@Override
					public StandardEvent map(String value) throws Exception {
						StandardEvent standardEvent = StandardEvent.parse(value);
						return standardEvent;
					}
				})
				.assignTimestampsAndWatermarks(
						new AssignerWithPeriodicWatermarks<StandardEvent>() {
							Long currentMaxTimestamp = 0L;
							Long maxDelayTime = 5000L;

							@Override
							public long extractTimestamp(StandardEvent s, long l) {
								currentMaxTimestamp = Timestamp.valueOf(Utils.transforDate(s.getField("CREATE_TIME"))).getTime();
								return currentMaxTimestamp;
							}

							@Nullable
							@Override
							public Watermark getCurrentWatermark() {
								long time = currentMaxTimestamp - maxDelayTime;
								//logger.info("getCurrentWatermark.............."+time);
								return new Watermark(time);
							}
						}
				)
				.keyBy(new KeySelector<StandardEvent, Long>() {
					@Override
					public Long getKey(StandardEvent event) throws Exception {
						return Long.valueOf(event.getField("SNOW_ID"));
					}
				})
				.connect(confBroadcast)
				.process(
						new dealStreamProcessFunction()
				).setParallelism(9);
		DataStream<BeanField> kafkaData1 = kafkaData.map(new dealMapFunction(ruleParse)).setParallelism(6);
//		kafkaData1.print();
		Table inputTable = blinkStreamTableEnv.fromDataStream(kafkaData1,
				Schema.newBuilder()
						.column("snowId","STRING")
						.column("account","STRING")
						.column("action","INTEGER")
						.column("actionDesc","STRING")
						.column("assetIp","STRING")
						.column("attackStage","STRING")
						.column("ausIndex","STRING")
						.column("averageByteFlow","INTEGER")
						.column("averagePackageFlow","INTEGER")
						.column("baseline","STRING")
						.column("bizId","STRING")
						.column("bizName","STRING")
						.column("confidence","INTEGER")
						.column("createTime","TIMESTAMP_LTZ(3)")
						.column("databaseName","STRING")
						.column("dataType","STRING")
						.column("deviceIp","STRING")
						.column("deviceName","STRING")
						.column("deviceParentType","STRING")
						.column("deviceType","STRING")
						.column("direction","STRING")
						.column("directionDesc","STRING")
						.column("downBaseLineFlow","INTEGER")
						.column("downFlowTotal","INTEGER")
						.column("dstAssetId","INTEGER")
						.column("dstAssetGroup","STRING")
						.column("dstAssetKey","INTEGER")
						.column("dstAssetName","STRING")
						.column("dstAssetPublic","INTEGER")
						.column("dstAssetStatus","INTEGER")
						.column("dstAssetSubType","STRING")
						.column("dstAssetType","INTEGER")
						.column("dstBizId","STRING")
						.column("dstCity","STRING")
						.column("dstCountry","STRING")
						.column("dstDomainName","STRING")
						.column("dstIntelDesc","STRING")
						.column("dstIntelId","STRING")
						.column("dstIntelType","INTEGER")
						.column("dstIp","STRING")
						.column("dstLatitude","STRING")
						.column("dstLongitude","STRING")
						.column("dstOrgId","STRING")
						.column("dstOrgName","STRING")
						.column("dstPort","STRING")
						.column("dstPost","STRING")
						.column("dstProvince","STRING")
						.column("dstSubDomainName","STRING")
						.column("eventName","STRING")
						.column("eventOneType","INTEGER")
						.column("eventThreeType","INTEGER")
						.column("eventTwoType","INTEGER")
						.column("eventThreeTypeDesc","STRING")
						.column("eventOneTypeDesc","STRING")
						.column("eventTwoTypeDesc","STRING")
						.column("eventType","STRING")
						.column("extAttr","STRING")
						.column("fileHash","STRING")
						.column("fileName","STRING")
						.column("filePath","STRING")
						.column("idCard","STRING")
						.column("infectionFile","STRING")
						.column("insertTime","TIMESTAMP_LTZ(3)")
						.column("installNum","INTEGER")
						.column("intelId","STRING")
						.column("intelType","STRING")
						.column("logSubType","STRING")
						.column("logType","STRING")
						.column("mailFileSize","STRING")
						.column("mailNum","INTEGER")
						.column("mailProtocol","STRING")
						.column("mailRecipient","STRING")
						.column("mailSender","STRING")
						.column("mailType","STRING")
						.column("mailTypeDesc","STRING")
						.column("mainAccount","STRING")
						.column("malwareName","STRING")
						.column("malwareSubType","STRING")
						.column("malwareType","STRING")
						.column("mobile","STRING")
						.column("msg","STRING")
						.column("name","STRING")
						.column("operateContent","STRING")
						.column("operateType","STRING")
						.column("orgId","INTEGER")
						.column("orgPath","STRING")
						.column("password","STRING")
						.column("payload","STRING")
						.column("pcap","STRING")
						.column("peakByteFlow","INTEGER")
						.column("peakPackageFlow","INTEGER")
						.column("percentBaseLineFlow","STRING")
						.column("percentFlowTotal","STRING")
						.column("policy","STRING")
						.column("position","STRING")
						.column("protocol","STRING")
						.column("rawMsg","STRING")
						.column("registerNum","INTEGER")
						.column("responseCode","STRING")
						.column("result","INTEGER")
						.column("riskLevel","INTEGER")
						.column("riskLevelDesc","STRING")
						.column("sceneId","INTEGER")
						.column("source","INTEGER")
						.column("sourceEventThreeType","INTEGER")
						.column("sourceEventThreeTypeDesc","STRING")
						.column("sourceType","INTEGER")
						.column("srcAssetId","INTEGER")
						.column("srcAssetGroup","STRING")
						.column("srcAssetKey","INTEGER")
						.column("srcAssetName","STRING")
						.column("srcAssetPublic","INTEGER")
						.column("srcAssetStatus","INTEGER")
						.column("srcAssetSubType","STRING")
						.column("srcAssetType","STRING")
						.column("srcBizId","STRING")
						.column("srcCity","STRING")
						.column("srcCountry","STRING")
						.column("srcDomainName","STRING")
						.column("srcDomain2Name","STRING")
						.column("srcIntelDesc","STRING")
						.column("srcIntelId","STRING")
						.column("srcIntelType","INTEGER")
						.column("srcIp","STRING")
						.column("srcLatitude","STRING")
						.column("srcLongitude","STRING")
						.column("srcOrgId","STRING")
						.column("srcOrgName","STRING")
						.column("srcPort","STRING")
						.column("srcPost","STRING")
						.column("srcProvince","STRING")
						.column("srcSubDomainName","STRING")
						.column("subAccount","STRING")
						.column("tabelName","STRING")
						.column("tags","STRING")
						.column("tenantId","INTEGER")
						.column("terminalNum","INTEGER")
						.column("threatType","STRING")
						.column("totalByteFlow","INTEGER")
						.column("totalPackageFlow","INTEGER")
						.column("type","STRING")
						.column("upBaseLineFlow","INTEGER")
						.column("upFlowTotal","INTEGER")
						.column("url","STRING")
						.column("user","STRING")
						.column("userName","STRING")
						.column("userOrgName","STRING")
						.column("userType","STRING")
						.column("vulnId","STRING")
						.column("vulnInfo","STRING")
						.column("vulnLevel","STRING")
						.column("alertSignatureIdL","STRING")
						.column("vulnType","STRING")
						.column("alertInfo","STRING")
						.column("domain","STRING")
						.column("flag","STRING")
						.column("flow","INTEGER")
						.column("flowUp","INTEGER")
						.column("flowDown","INTEGER")
						.column("command","STRING")
						.watermark("createTime", "SOURCE_WATERMARK()")
						.build());
		blinkStreamTableEnv.createTemporaryView("InputTable", inputTable);
//		Table resultTable = blinkStreamTableEnv.sqlQuery("SELECT * FROM InputTable");
//		blinkStreamTableEnv.toRetractStream(resultTable, BeanField.class).print("query==");

		Table resultTableIds = blinkStreamTableEnv.sqlQuery("SELECT deviceType,count(1) FROM InputTable where deviceType = 'IDS' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType");
		blinkStreamTableEnv.toRetractStream(resultTableIds, Row.class).print("queryIds==");

		Table resultTableTda = blinkStreamTableEnv.sqlQuery("SELECT deviceType,count(1) FROM InputTable where deviceType = 'TDA' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType");
		blinkStreamTableEnv.toRetractStream(resultTableTda, Row.class).print("queryTda==");

		Table resultTableIps= blinkStreamTableEnv.sqlQuery("SELECT deviceType,count(1) FROM InputTable where deviceType = 'IPS' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType");
		blinkStreamTableEnv.toRetractStream(resultTableIps, Row.class).print("queryIps==");


		String querySQL4 = "select b.* from (select deviceType,snowId ,count(1) as cnt from TABLE(\n" +
				"TUMBLE(TABLE InputTable, DESCRIPTOR(createTime), INTERVAL '10' MINUTES))" +
				" where deviceType='TDA' GROUP BY window_start, window_end,deviceType,snowId having count(1)=1) a,(SELECT * FROM InputTable) b" +
				" where a.snowId =b.snowId ";
		 resultTableIps= blinkStreamTableEnv.sqlQuery(querySQL4);
		blinkStreamTableEnv.toRetractStream(resultTableIps, BeanField.class).print("BeanField==");

//		String sql="SELECT * FROM (SELECT deviceType,count(1) FROM InputTable where deviceType = 'IDS' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType)) a,(SELECT deviceType,count(1) FROM InputTable where deviceType = 'TDA' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceType) b";
//		Table resultTableIps= blinkStreamTableEnv.sqlQuery(sql);
//		blinkStreamTableEnv.toRetractStream(resultTableIps, Row.class).print("queryIps==");

//		blinkStreamTableEnv.createTemporaryView("table1", resultTableIds);
//		blinkStreamTableEnv.createTemporaryView("table2", resultTableTda);
//		blinkStreamTableEnv.createTemporaryView("table3", resultTableIps);
//		String querySQL8 = "select * from table2 ,table1,table3 where table1.srcIp=table2.srcIp and table1.dstIp=table3.dstIp";
//		Table table8 = blinkStreamTableEnv.sqlQuery(querySQL8);
//		blinkStreamTableEnv.toRetractStream(table8, Row.class).print("table8==");

		env.execute("Broadcast test kafka");
	}

}
